package hadoop.mapreduce.partition;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * 驱动类
 */
public class PhoneDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration entries = new Configuration();
        Job job = Job.getInstance(entries);

        job.setJarByClass(PhoneDriver.class);
        job.setMapperClass(PhoneMapper.class);
        job.setReducerClass(PhoneReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(PhoneBean.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //自定义分区类
        job.setPartitionerClass(ProvincePartitioner.class);
        //自定义reduce task 数量X,数量要与ProvincePartitioner类中定义的分区数量Y相符
        //若X=1怎分区无效,只会出现一个文件
        //若X<Y && X!=1 则报错
        //若X=Y,正常
        //若X>Y,正常,会有空文件
        job.setNumReduceTasks(5);

        FileInputFormat.setInputPaths(job,new Path("C:\\tmp\\phone_data.txt"));
        FileOutputFormat.setOutputPath(job,new Path("C:\\tmp\\aa"));
//        FileInputFormat.setInputPaths(job, new Path(args[0]));
//        FileOutputFormat.setOutputPath(job,new Path(args[1]));

        boolean result = job.waitForCompletion(true);
        System.exit(result?0:1);
    }
}
